package cn.bud.msgbus.consumer.client.provider;

import com.zl9c.general.bus.bean.BusURL;
import com.zl9c.general.bus.bean.MsgBody;
import com.zl9c.general.bus.server.MsgBusDataPusher;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.web.ServerProperties;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.http.*;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * ProviderHandler
 *
 * @author hxj
 * @version 1.0
 * @description
 * @date 2024-07-12 10:14
 */
@Slf4j
@Component
public class ProviderHandler implements InvocationHandler {

    private static Map<Object, ProviderProxy> proxyMap = new ConcurrentHashMap<>();

    @Resource
    private MsgBusDataPusher msgBusDataPusher;
    @Resource
    private RestTemplate restTemplate;
    @Resource
    private ServerProperties serverProperties;

    public void addProxy(ProviderProxy proxy) {
        synchronized (proxyMap) {
            proxyMap.put(proxy.proxyObj.getClass().getSimpleName(), proxy);
        }
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
//        if (Object.class.getDeclaringClass().equals(method.getDeclaringClass())) {
//            return method.invoke(proxy, args);
//        }
        // 在这里编写自定义的实现逻辑
        MsgBody msg = new MsgBody();
        SubscribeProvide spAnno = AnnotationUtils.findAnnotation(method, SubscribeProvide.class);
        if (spAnno == null) {
            return null;
        }
        msg.setTopic(spAnno.topic());
        msg.setAsync(spAnno.async());
        for (Object arg : args) {
            msg.setData(arg);
        }
        // 推送数据
        ProviderProxy providerProxy = proxyMap.get(proxy.getClass().getSimpleName());
        BusURL url = providerProxy.getBusUrl();
        if (url.isLocalServer(serverProperties.getPort())) {
            msgBusDataPusher.pushMsg(msg);
        }else if (url.isHttp()) {
            httpPost(msg.jsonObj().toJSONString(0), url.getUrl(), 0);
        }else {
            log.info("推送数据：{}", msg);
        }
        return "push success";
    }


    private void httpPost(String jsonStr, String url, int retryTimes) {
        if (restTemplate == null) {
            log.warn("http发布订阅消息异常，未找到restTemplate实例");
            return;
        }
        try {
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
            HttpEntity<String> httpEntity = new HttpEntity<>(jsonStr,headers);

            ResponseEntity<Object> rsp = restTemplate.postForEntity(url,httpEntity, Object.class);
            if (rsp.getStatusCode() == HttpStatus.OK) {
                log.debug("http发布订阅消息：url={}，推送的数据：{}",
                        url, jsonStr);
            }
        }catch (Exception e) {
            retryTimes++;
            log.warn("http发布订阅消息异常：url={}，第{}次推送，推送的数据：{}，异常提示：{}",
                    url, retryTimes, jsonStr, e.getMessage());
            if (retryTimes < 3) {
                httpPost(jsonStr, url, retryTimes);
            }
            throw e;
        }
    }

}
